﻿using Confluent.Kafka;
using EventBus.Common;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace EventBus.Kafka
{
    /// <summary>
    /// kafka发布者
    /// </summary>
    public class KafkaPublisher : IKafkaPublisher
    {
        private readonly IkafkaConnection _kafkaConnection;
        public KafkaPublisher(IkafkaConnection kafkaConnection)
        {
            _kafkaConnection = kafkaConnection;
        }
        public async Task PublishAsync(string topic, object content)
        {
            try
            {
                var producer = _kafkaConnection.GetProducer();
                var message = SwifterJsonSerializer.SerializeObject(content);
                var messageKey = Guid.NewGuid().ToString();
                var messageValue = Encoding.Default.GetBytes(message);
                await producer.ProduceAsync(topic, new Message<string, byte[]> { Key = messageKey, Value = messageValue }).ContinueWith(e =>
                       {
                           switch (e.Result.Status)
                           {
                               case PersistenceStatus.NotPersisted:
                                   Console.WriteLine($"{e.Result.Topic}{content}发送失败。持久化失败!");
                                   break;
                               case PersistenceStatus.PossiblyPersisted:
                                   Console.WriteLine($"{e.Result.Topic}{content}发送成功。持久化还没有成功!");
                                   break;
                               case PersistenceStatus.Persisted:
                                   Console.WriteLine($"{e.Result.Topic}{content}持久化成功!");
                                   break;
                               default:
                                   Console.WriteLine($"{e.Result.Topic}{content}持久化异常!");
                                   break;
                           }
                       });

                producer.Flush(TimeSpan.FromSeconds(10));

            }
            catch (Exception)
            {

                throw;
            }
        }
    }
}
